package org.tao.lightningmq.common.remote;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.common.dto.*;
import org.tao.lightningmq.common.enums.BrokerEventCode;
import org.tao.lightningmq.common.enums.NameServerResponseCode;
import org.tao.lightningmq.common.tcp.TcpMsg;

/**
 * @Author lht
 * @date 2025/1/8 - 17:32
 * @description:
 */
public class NameServerRemoteRespHandler extends SimpleChannelInboundHandler<TcpMsg> {

    private static final Logger LOGGER = LoggerFactory.getLogger(NameServerRemoteRespHandler.class);

    private BrokerMasterChangeCallBack brokerMasterChangeCallBack;
    private NameserverLogoutCallBack nameserverLogoutCallBack;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TcpMsg tcpMsg) throws Exception {
        int code = tcpMsg.getCode();
        byte[] body = tcpMsg.getBody();
        BaseNameServerRemoteDTO respDTO = null;
        if (NameServerResponseCode.REGISTRY_SUCCESS.getCode() == code) {
            respDTO = JSON.parseObject(body, ServiceRegistryRespDTO.class);
        } else if (NameServerResponseCode.ERROR_USER_OR_PASSWORD.getCode() == code) {
            respDTO = JSON.parseObject(body, ServiceRegistryRespDTO.class);
        } else if (NameServerResponseCode.HEART_BEAT_SUCCESS.getCode() == code) {
            respDTO = JSON.parseObject(body, HeartBeatDTO.class);
        } else if (NameServerResponseCode.PULL_BROKER_ADDRESS_SUCCESS.getCode() == code) {
            respDTO = JSON.parseObject(body, PullBrokerIpRespDTO.class);
        } else if (BrokerEventCode.BROKER_MASTER_CHANGE_MSG.getCode() == code) {
            BrokerMasterChangeReqDTO reqDTO = JSON.parseObject(body, BrokerMasterChangeReqDTO.class);
            brokerMasterChangeCallBack.onReceive(reqDTO);
            return;
        } else {
            ctx.close();
            throw new IllegalAccessException("error code");
        }
        setMsg(tcpMsg, respDTO);
    }

    private static void setMsg(TcpMsg tcpMsg, BaseNameServerRemoteDTO respDTO) {
        SyncFuture syncFuture = SyncFutureManager.get(respDTO.getMsgId());
        if (syncFuture != null) {
            syncFuture.setResp(tcpMsg);
        }
    }

    public void setBrokerMasterChangeCallBack(BrokerMasterChangeCallBack brokerMasterChangeCallBack) {
        this.brokerMasterChangeCallBack = brokerMasterChangeCallBack;
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ctx.close();
        new Thread(nameserverLogoutCallBack::nsLogoutCallBack).start();
    }

    public void setNameServerLogoutCallBack(NameserverLogoutCallBack nameServerLogoutCallBack) {
        this.nameserverLogoutCallBack = nameServerLogoutCallBack;
    }
}
